package com.zsy.rocketdemo.produce;

import lombok.Getter;
import lombok.Setter;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.stream.IntStream;

@Setter
@Getter
public abstract class BaseProducer {

    protected DefaultMQProducer producer;
    private String topic = "zsyTopicTest";
    protected String tag;

    @PostConstruct
    public void init() throws MQClientException {
        this.producer = new DefaultMQProducer(getTag());
        this.producer.setNamesrvAddr("47.100.22.19:9876");
        this.producer.setInstanceName(this.tag);
        this.producer.start();
        System.out.println(this.tag + "Producer启动。");
    }

    public void sendMessage(int num) {
        IntStream.range(0, num).forEach(i -> {
            try {
                sendMessage(buildMessage(i));
            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        });
    }

    public void sendMessage(Message message) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        this.producer.send(message);
    }

    public Message buildMessage(int num) throws UnsupportedEncodingException {
        return new Message(getTopic(),
                getTag(),
                ("zsy Hello RocketMQ " + getTag() + " " + num).getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
    }

}
